package day04.actorPlus


import java.util.UUID

import akka.actor.{Actor, ActorRef, ActorSelection, ActorSystem, Props}
import akka.actor.Actor.Receive
import com.typesafe.config.ConfigFactory
import scala.concurrent.duration._

import scala.language.postfixOps

/**
  * Created by root on 2018/11/26.
  */
class Worker(var workerHost:String , var workerPort:Int , var memory:Int,var cores:Int) extends Actor{

  val workerId = UUID.randomUUID().toString
  var masterRef: ActorSelection = _
  /**
    * preStart 在构造器之后，receive方法执行之前执行一次
    */
  override def preStart(): Unit = {
    //让worker 跟 master 建立链接，并发送给master一个小厮
    masterRef = context.actorSelection(s"akka.tcp://${Master.MASTER_SYSTEM}@$workerHost:$workerPort/user/${Master.MASTER_NAME}")

    masterRef ! RegisterWorker(workerId,memory,cores)

  }

  override def receive: Receive = {
    //注册成功后的收到的消息
    case RegisteredWorker =>{
      //启动定时器，定期向master发送心跳
      //导入一个定时器依赖的隐式转换
      import context.dispatcher    //?
      context.system.scheduler.schedule(0 millis,10000 millis,self,SendHeartBeat)

    }

    case SendHeartBeat => {
      //给master 发送心跳
      masterRef.!(HeartBeat(workerId))
    }
  }
}

object Worker{

  val WORKER_SYSTEM = "Worker_system"
  val WORKER_NAME = "Worker"

  def main(args: Array[String]): Unit = {

    //master args
    val masterHost = "127.0.0.1"
    val masterPort = 8888

    // worker
    val workerHost = "127.0.0.1"
    val workerPort = 9999
    val memory = 1024
    val cores = 2

    val confStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = $workerHost
         |akka.remote.netty.tcp.port = $workerPort
      """.stripMargin

    val config = ConfigFactory.parseString(confStr)
    // 创建 actorSystem(单例)
    val actorSystem = ActorSystem(WORKER_SYSTEM,config)

    //通过创建的actorSystem 在创建actor    ,actor创建会执行声明周期方法
    val workerActor: ActorRef = actorSystem.actorOf(Props(new Worker(workerHost,workerPort,memory,cores)),WORKER_NAME)




  }
}

